package demo.kafka.stream.config;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

public class StreamsConfigFactory {

    public static Properties createConfigFromFile(String configFilePath) throws IOException {
        Properties props = new Properties();
        try (InputStream input = StreamsConfigFactory.class.getClassLoader().getResourceAsStream(configFilePath)) {
            if (input == null) {
                throw new ConfigException("Configuration file not found: " + configFilePath);
            }
            props.load(input);
        }

        Properties streamsConfig = new Properties();

        // Required settings
        streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, props.getProperty("kafka.application.id"));
        streamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, props.getProperty("kafka.bootstrap.servers"));
        streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // Optional: Exactly-Once processing
        String processingGuarantee = props.getProperty("kafka.processing.guarantee");
        if (processingGuarantee != null && !processingGuarantee.trim().isEmpty()) {
            streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee);
        }

        // Optional: Exception handler
        String exceptionHandler = props.getProperty("kafka.default.deserialization.exception.handler");
        if (exceptionHandler != null && !exceptionHandler.trim().isEmpty()) {
            streamsConfig.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, exceptionHandler);
        }

        // Optional: Security settings (example for SASL/PLAIN)
        if ("SASL_PLAINTEXT".equals(props.getProperty("kafka.security.protocol"))) {
            streamsConfig.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
            streamsConfig.put("sasl.mechanism", "PLAIN");
            streamsConfig.put("sasl.jaas.config", props.getProperty("kafka.sasl.jaas.config"));
        }

        // Recommended performance settings
        streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // For development
        // streamsConfig.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0"); // Disable cache for development to see immediate results

        return streamsConfig;
    }

    public static String getInputTopic(Properties appProps) {
        return appProps.getProperty("kafka.input.topic");
    }

    public static String getOutputTopic(Properties appProps) {
        return appProps.getProperty("kafka.output.topic");
    }
}
